package com.example.mq.mqclient;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.Consumer;
import com.example.mq.common.MqException;
import com.example.mq.common.net.*;
import com.example.mq.mqserver.core.BasicProperties;
import com.example.mq.mqserver.core.ExchangeType;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class Channel {

    private String channelId;

    //记录当前channel所属的连接
    private Connection connection;

    //用来存储 服务器、客户端之间的响应信息<rid,BasicReturns>
    private ConcurrentHashMap<String, BasicReturns> basicReturnMap = new ConcurrentHashMap<String, BasicReturns>();

    //如果当前channel订阅了队列，需要记录回调
    //此处约定一个channel只能有一个回调
    private Consumer consumer;

    public Channel(String channelId, Connection connection) {
        this.channelId = channelId;
        this.connection = connection;
    }


    public String getChannelId() {
        return channelId;
    }

    public void setChannelId(String channelId) {
        this.channelId = channelId;
    }

    public Connection getConnection() {
        return connection;
    }

    public void setConnection(Connection connection) {
        this.connection = connection;
    }


    public ConcurrentHashMap<String, BasicReturns> getChannels() {
        return basicReturnMap;
    }

    public void setChannels(ConcurrentHashMap<String, BasicReturns> channels) {
        this.basicReturnMap = channels;
    }

    public Consumer getConsumer() {
        return consumer;
    }

    public void setConsumer(Consumer consumer) {
        this.consumer = consumer;
    }

    //当putRetruns rid,BasicReturns消息后，会唤醒当前对象（channel）所有线程，尝试去获取basicRetruenMap中的元素，获取成功，
    // 返回这个basicReturns（最后对应的put了的元素也要remvoe掉）
    private BasicReturns waitResult(String rid) throws InterruptedException {
        BasicReturns basicReturns = null;
        while ((basicReturns = basicReturnMap.get(rid)) == null) {
            //包裹还没来如果null，进行阻塞等待
            synchronized (this) {

                //当前对象进入线程等待
                wait();//notify随机唤醒当前对象的一个线程 notifyAll唤醒所有当前对象的线程
            }
        }

        basicReturnMap.remove(rid);//结果收到之后，从集合中删除

        return basicReturns;
    }

    //把服务器响应信息放到 basicReturnMap中，同时提醒当前对象的线程去获取，把响应返回给客户端
    public void putReturns(BasicReturns basicReturns) {
        basicReturnMap.put(basicReturns.getRid(), basicReturns);
        // 当前也不知道有多少个线程在等待上述的这个响应.
        // 把所有的等待的线程都唤醒.
        synchronized (this) {
            this.notifyAll();
        }
    }

    //告诉服务器创建channel
    public boolean createChannel() throws IOException, InterruptedException {
        Request request = new Request();

        BasicArguments args = new BasicArguments();
        args.setChannelId(channelId);
        args.setRid(generateRid());
        args.setVirtualHostName(connection.getVirtualHostName());
        byte[] payLoad = BinaryTool.toBytes(args);

        request.setType(0x1);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);

        connection.writeRequest(request);

        //等待服务器响应
        BasicReturns basicReturns = waitResult(args.getRid());

        return basicReturns.isOk();
    }



    private String generateRid() {
        return "Rid-" + UUID.randomUUID();
    }

    //关闭channel
    public boolean close() throws IOException, InterruptedException {

        //创建数据
        BasicArguments args = new BasicArguments();
        args.setRid(generateRid());
        args.setChannelId(channelId);
        args.setVirtualHostName(connection.getVirtualHostName());
        byte[] payLoad = BinaryTool.toBytes(args);

        //写入数据
        Request request = new Request();
        request.setType(0x2);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);
        connection.writeRequest(request);

        //等待响应
        BasicReturns basicReturns = waitResult(args.getRid());

        //返回结果
        return basicReturns.isOk();
    }

    //创建交换机
    public boolean exchangeDeclare(String exchangeName
            , ExchangeType exchangeType
            , boolean durable
            , boolean autoDelete
            , Map<String, Object> arguments) throws IOException, InterruptedException {

        //生成交换机类
        ExchangeDeclareArguments arg = new ExchangeDeclareArguments();

        arg.setRid(generateRid());
        arg.setChannelId(channelId);
        arg.setVirtualHostName(connection.getVirtualHostName());

        arg.setExchangeName(exchangeName);
        arg.setExchangeType(exchangeType);
        arg.setDurable(durable);
        arg.setAutoDelete(autoDelete);
        arg.setArguments(arguments);


        byte[] payLoad = BinaryTool.toBytes(arg);
        Request request = new Request();
        request.setType(0x3);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);

        //写入请求
        connection.writeRequest(request);

        BasicReturns basicReturns = waitResult(arg.getRid());
        return basicReturns.isOk();
    }

    //删除交换机
    public boolean exchangeDelete(String exchangeName) throws IOException, InterruptedException {
        ExchangeDeleteArguments arguments = new ExchangeDeleteArguments();

        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setVirtualHostName(connection.getVirtualHostName());
        arguments.setExchangeName(exchangeName);
        byte[] payLoad = BinaryTool.toBytes(arguments);


        Request request = new Request();
        request.setType(0x4);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);
        connection.writeRequest(request);
        BasicReturns returns = waitResult(arguments.getRid());

        return returns.isOk();
    }

    //创建队列
    public boolean queueDeclare(
            String queueName
            , boolean durable
            , boolean exclusive
            , boolean autoDelete
            , Map<String, Object> arguments) throws IOException, InterruptedException {

        //生成队列参数
        QueueDeclareArguments args = new QueueDeclareArguments();
        args.setRid(generateRid());
        args.setChannelId(channelId);
        args.setVirtualHostName(connection.getVirtualHostName());

        args.setQueueName(queueName);
        args.setDurable(durable);
        args.setExclusive(exclusive);
        args.setAutoDelete(autoDelete);
        args.setArguments(arguments);

        byte[] payLoad = BinaryTool.toBytes(args);

        //参数写入请求
        Request request = new Request();
        request.setType(0x5);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);

        //发送请求
        connection.writeRequest(request);

        //等待响应
        BasicReturns returns = waitResult(args.getRid());

        return returns.isOk();

    }

    //删除队列
    public boolean queueDelete(String queueName) throws IOException, InterruptedException {
        QueueDeleteArguments arguments = new QueueDeleteArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setVirtualHostName(connection.getVirtualHostName());
        arguments.setQueueName(queueName);
        byte[] payLoad = BinaryTool.toBytes(arguments);

        Request request = new Request();
        request.setType(0x6);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);

        connection.writeRequest(request);

        BasicReturns returns = waitResult(arguments.getRid());
        return returns.isOk();
    }

    //队列绑定
    public boolean queueBind(String queueName, String exchangeName, String bindingKey) throws IOException, InterruptedException {
        QueueBindArguments arguments = new QueueBindArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setVirtualHostName(connection.getVirtualHostName());
        arguments.setQueueName(queueName);
        arguments.setExchangeName(exchangeName);
        arguments.setBindingKey(bindingKey);

        byte[] payLoad = BinaryTool.toBytes(arguments);

        Request request = new Request();
        request.setType(0x7);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);

        connection.writeRequest(request);

        BasicReturns returns = waitResult(arguments.getRid());
        return returns.isOk();
    }

    //解绑
    public boolean queueUnbind(String queueName, String exchangeName) throws IOException, InterruptedException {

        QueueUnbindArguments arguments = new QueueUnbindArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setVirtualHostName(connection.getVirtualHostName());
        arguments.setQueueName(queueName);
        arguments.setExchangeName(exchangeName);

        byte[] payLoad = BinaryTool.toBytes(arguments);

        Request request = new Request();
        request.setType(0x8);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);

        connection.writeRequest(request);

        BasicReturns returns = waitResult(arguments.getRid());
        return returns.isOk();
    }

    //发布消息
    public boolean basicPublish(String exchangeName
            , String routingKey
            , BasicProperties basicProperties
            , byte[] body) throws IOException, InterruptedException {

        BasicPublishArguments args = new BasicPublishArguments();

        args.setRid(generateRid());
        args.setChannelId(channelId);
        args.setVirtualHostName(connection.getVirtualHostName());
        args.setExchangeName(exchangeName);
        args.setRoutingKey(routingKey);
        args.setProperties(basicProperties);
        args.setBody(body);

        byte[] payLoad = BinaryTool.toBytes(args);

        Request request = new Request();

        request.setType(0x9);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);
        connection.writeRequest(request);

        BasicReturns returns = waitResult(args.getRid());

        return returns.isOk();

    }

    //消费者回调
    public boolean  basicConsume(String queueName
            , boolean autoDelete
            , Consumer consumer) throws IOException, InterruptedException {
        this.consumer = consumer;

        BasicConsumeArguments arguments = new BasicConsumeArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setVirtualHostName(connection.getVirtualHostName());
        //消费者标识用channelId表示
        arguments.setConsumerTag(channelId);
        arguments.setQueueName(queueName);
        arguments.setAutoAck(autoDelete);
        byte[] payLoad = BinaryTool.toBytes(arguments);

        Request request = new Request();
        request.setType(0xA);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);
        connection.writeRequest(request);

        BasicReturns returns = waitResult(arguments.getRid());
        return returns.isOk();
    }

    //发送basicAck确认
    public boolean basicAck(String queueName, String messageId) throws IOException, InterruptedException {
        BasicAckArguments arguments = new BasicAckArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setVirtualHostName(connection.getVirtualHostName());
        arguments.setQueueName(queueName);
        arguments.setMessageId(messageId);

        byte[] payLoad = BinaryTool.toBytes(arguments);

        Request request = new Request();

        request.setType(0xB);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);
        connection.writeRequest(request);

        BasicReturns returns = waitResult(arguments.getRid());
        return returns.isOk();
    }
    
    // 创建虚拟主机
    public boolean createVirtualHost(String virtualHostName) throws IOException, InterruptedException {
        VirtualHostArguments arguments = new VirtualHostArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setVirtualHostName(virtualHostName);
        
        byte[] payLoad = BinaryTool.toBytes(arguments);
        
        Request request = new Request();
        request.setType(0xD);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);
        connection.writeRequest(request);
        
        BasicReturns returns = waitResult(arguments.getRid());
        return returns.isOk();
    }
    
    // 删除虚拟主机
    public boolean deleteVirtualHost(String virtualHostName) throws IOException, InterruptedException {
        VirtualHostArguments arguments = new VirtualHostArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setVirtualHostName(virtualHostName);
        
        byte[] payLoad = BinaryTool.toBytes(arguments);
        
        Request request = new Request();
        request.setType(0xE);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);
        connection.writeRequest(request);
        
        BasicReturns returns = waitResult(arguments.getRid());
        return returns.isOk();
    }
    
    // 获取所有虚拟主机列表
    public List<String> getVirtualHostList() throws IOException, InterruptedException {
        BasicArguments arguments = new BasicArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        
        byte[] payLoad = BinaryTool.toBytes(arguments);
        
        Request request = new Request();
        request.setType(0xF);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);
        connection.writeRequest(request);
        
        VirtualHostListReturns returns = (VirtualHostListReturns) waitResult(arguments.getRid());
        return returns.getVirtualHostNames();
    }
}
